package com.fwmagic.spark.core.cases.threadsafe

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object TaskThreadSafe1 {
    def main(args: Array[String]): Unit = {
        val isLocal = args(0).toBoolean

        val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName)
        if (isLocal) {
            conf.setMaster("local[*]")
        }

        val sc = new SparkContext(conf)

        val lines: RDD[String] = sc.textFile(args(1))

        /**
          * DateUtils对象在Driver端创建
          * 1.DateUtils类需要实现序列化才可被发送到Executor端
          * 2.DateUtils是object单例的，在Executor中所有的Task共享一份，会有线程安全的问题
          * 3.方法1：
          *     想要线程安全就需要在DateUtils的getTime方法加上synchronized同步
          *   方法2：
          *     使用一个线程安全的时间格式化类，而不是SimpleDateFormat
          */
        val dateUtils = DateUtils

        val mapRDD: RDD[(String, Long)] = lines.map(line => {
            val time: Long = dateUtils.getTime2(line)
            (line, time)
        })

        val tuples: Array[(String, Long)] = mapRDD.collect()

        tuples.foreach(println)

        sc.stop()
    }
}
